package rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 负责监听队列消息 消费处理
 */
public class Consumer {
    //连接客户端
    private Channel channel;

    /**
     * @Before的案例方法 是在所有的单元测试test方法之前执行
     */
    @Before
    public void init() throws IOException, TimeoutException {
        //通过java客户端建立和rabbitmq的连接
        ConnectionFactory factory=new ConnectionFactory();
        //给工厂提供链接属性 ip:port
        factory.setHost("192.168.8.100");
        factory.setPort(5672);
        //权限账号 admin/123456
        factory.setUsername("admin");
        factory.setPassword("123456");
        //虚拟主机,每隔连接工厂,由一个程序管理,程序所属的环境 连接的虚拟主机 virtual_host不同
        //实现不同环境的数据隔离 默认就是/
        factory.setVirtualHost("/");
        Connection connection = factory.newConnection();
        //可以从长连接获取短链接
        channel = connection.createChannel();
    }
    @Test
    public void consumer01() throws IOException {
        //为消息消费处理准备一个回调对象deliverCallback;
        DeliverCallback deliver=new DeliverCallback() {
            //消费处理方法handle

            /**
             *
             * @param consumerTag 处理消息的消费者标识
             * @param message 消息队列传递过来的消息对象 包含发送的消息所有数据
             *
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("消费者1接收到消息");
                //从message中获取 消息内容
                byte[] body = message.getBody();
                String msg=new String(body);
                //消息来源交换机 消息路由键
                message.getEnvelope().getExchange();
                message.getEnvelope().getRoutingKey();
                //消息携带的属性 现在是null
                message.getProperties();
                System.out.println("收到消息数据:"+msg);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            /**
             消费者 停止监听 拒绝消息 断开连接 调用的方法
             一般是在消费者 停止消费的时候回调 释放一些消费者持有的资源
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("消费者1停止监听");
            }
        };
        //调用客户端消费方法
        /**
         * String queue: 队列名称,当前连接 对象监听消费的队列是谁
         * boolean autoAck: 是否自动确认 true自动确认 false手动确认
         * 消费方法 处理是异步线程,消费结果 消费状态 异步回调.
         */
        channel.basicConsume("queue02",true,deliver,cancel);
        //保证回调 可以看到处理结果 线程阻塞等待 死循环
        while(true);
    }
    @Test
    public void consumer02() throws IOException {
        //为消息消费处理准备一个回调对象deliverCallback;
        DeliverCallback deliver=new DeliverCallback() {
            //消费处理方法handle

            /**
             *
             * @param consumerTag 处理消息的消费者标识
             * @param message 消息队列传递过来的消息对象 包含发送的消息所有数据
             *
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("消费者2接收到消息");
                //从message中获取 消息内容
                byte[] body = message.getBody();
                String msg=new String(body);
                //消息来源交换机 消息路由键
                message.getEnvelope().getExchange();
                message.getEnvelope().getRoutingKey();
                //消息携带的属性 现在是null
                message.getProperties();
                System.out.println("收到消息数据:"+msg);
                //手动确认 根据消费结果 确认 可以重新投递
                //channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                //手动确认 消费失败 重新投递
                /*channel.basicNack(
                        message.getEnvelope().getDeliveryTag(),
                        false,
                        true);*/
                //直接拒绝 不需重新投递 消息去哪了? 去了死信队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            /**
             消费者 停止监听 拒绝消息 断开连接 调用的方法
             一般是在消费者 停止消费的时候回调 释放一些消费者持有的资源
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("消费者1停止监听");
            }
        };
        //调用客户端消费方法
        /**
         * String queue: 队列名称,当前连接 对象监听消费的队列是谁
         * boolean autoAck: 是否自动确认 true自动确认 false手动确认
         * 消费方法 处理是异步线程,消费结果 消费状态 异步回调.
         */
        channel.basicConsume("queue01",false,deliver,cancel);
        //保证回调 可以看到处理结果 线程阻塞等待 死循环
        while(true);
    }
}
